package engine

import (
	"context"
	"sdn-executor/structs/queue"
	"sdn-executor/utils/erros"
	"sdn-executor/utils/routines"
	"time"
)

type Channel struct {
	que *queue.BlockingPriorityQueue[*Message]
	ctx context.Context
}

func NewChannel(ctx context.Context, maxSize int) *Channel {
	que, _ := queue.NewBlockingPriorityQueue[*Message](ctx, maxSize, queue.ClosePolicyClose)
	return &Channel{
		que: que,
		ctx: ctx,
	}
}

func (c *Channel) Start() {
	c.que.Start()
	routines.GoWithRecover("engineChannel", c.run)
}

func (c *Channel) Push(m *Message) *erros.Error {
	return c.que.Push(m)
}

func (c *Channel) PushWithTimeout(m *Message, timeout time.Duration) *erros.Error {
	return c.que.PushWithTimeout(m, timeout)
}

func (c *Channel) run() {
	for {
		select {
		case <-c.ctx.Done():
			return
		default:
			m, err := c.que.Pop()
			if err == nil {
				Execute(m)
			} else {
				// 已关闭
				return
			}

		}
	}
}
